package com.whale.api;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONPath;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.util.List;

public class _07_Transformation_Demos {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = env.readTextFile("flink_learn/data/transformation_input/");

        /**
         * map: 映射 类型lambda表达式的map
         */
//        source.map((MapFunction<String, UserInfo>) e -> JSONObject.parseObject(e, UserInfo.class)).print();

        /**
         * filter：类似lambda表达式的filter
         */
//        source.filter(e -> JSONObject.parseObject(e).getString("uid").equals("8")).print();

        /**
         * flatmap算子的演示
         *  把每个用户的好友信息，全部提取出来（带上用户自身的信息），并且压平，放入结果流中
         *  {"uid":1,"name":"zs","gender":"male","friends":[{"fid":2,"name":"aa"},{"fid":3,"name":"bb"}]}
         *  =>
         *  {"uid":1,"name":"zs","gender":"male","fid":2,"fname":"aa"}
         *  {"uid":1,"name":"zs","gender":"male","fid":3,"fname":"bb"}
         */
//        source
//                .map((MapFunction<String, UserInfo>) e -> JSONObject.parseObject(e, UserInfo.class))
//                .filter(e -> e.getUid() == 8) // 筛选出id=8的数据
//                .flatMap(new FlatMapFunction<UserInfo, UserFriendInfo>() {
//                    @Override
//                    public void flatMap(UserInfo e, Collector<UserFriendInfo> collector) throws Exception {
//                        List<FriendInfo> friends = e.getFriends();
//                        for (FriendInfo friend : friends) {
//                            UserFriendInfo userFriendInfo = new UserFriendInfo();
//                            userFriendInfo.setUid(e.getUid());
//                            userFriendInfo.setName(e.getName());
//                            userFriendInfo.setGender(e.getGender());
//                            userFriendInfo.setFid(friend.getFid());
//                            userFriendInfo.setFname(friend.getName());
//
//                            collector.collect(userFriendInfo);
//                        }
//                    }
//                })
//                .print();

        /**
         * keyBy
         */
//        source
//                .map((MapFunction<String, UserInfo>) e -> JSONObject.parseObject(e, UserInfo.class))
//                .flatMap((FlatMapFunction<UserInfo, UserFriendInfo>) (e, collector) -> {
//                    List<FriendInfo> friends = e.getFriends();
//                    for (FriendInfo friend : friends) {
//                        UserFriendInfo userFriendInfo = new UserFriendInfo();
//                        userFriendInfo.setUid(e.getUid());
//                        userFriendInfo.setName(e.getName());
//                        userFriendInfo.setGender(e.getGender());
//                        userFriendInfo.setFid(friend.getFid());
//                        userFriendInfo.setFname(friend.getName());
//
//                        collector.collect(userFriendInfo);
//                    }
//                })
//                .returns(new TypeHint<UserFriendInfo>() {}) // 使用lamdab表达式就得用 returns来表示参数类型
//                .map(userFriendInfo -> Tuple2.of(userFriendInfo.getGender(), 1)) // map-》把数据筛选成 <性别, 1>
//                .returns(new TypeHint<Tuple2<String, Integer>>() {})
//                .keyBy(tuple -> tuple.f0) // 根据性别分组
//                .sum(1) // 求和，为啥上边要把数量设置成1，方便统计sum
//                .print();

        /**
         * max / maxBy  都是滚动聚合：  算子不会把收到的所有数据全部攒起来；而是只在状态中记录上一次的聚合值，然后当新数据到达的时候，会根据逻辑去更新 状态中记录的聚合值，并输出最新状态数据
         * max / maxBy  区别： 更新状态的逻辑！  max只更新要求最大值的字段；  而 maxBy 会更新整条数据；
         * 同理，min和minBy也如此
         *
         * 比如select username, age, count(1) from user group by age
         * 一般，我们做group by 的时候，结果集里都是 分组的字段+聚合函数。 但是如果出现其他字段，那么这些字段是不准的
         * flink的max、min 同理，你看我下面的例子，可以翻译成 select uid, name, gender, friends from user group by gender
         * 那么uid, name, friends就是随机一个数据，但是如果是maxBy的话 则会返回gender最大的那个人的uid、name
         *
         * 这里有个小坑，如果我年龄30，A的年龄30，那么flink发现下一个数据没有比我大，则不会替换。 可以使用reduce解决
         */
//        source
//                .map((MapFunction<String, UserInfo>) e -> JSONObject.parseObject(e, UserInfo.class))
//                .map(bean -> Tuple4.of(bean.getUid(), bean.getName(), bean.getGender(), bean.getFriends().size()))
//                .returns(new TypeHint<Tuple4<Integer, String, String, Integer>>() {})
//                .keyBy(tp -> tp.f2)
//                .maxBy(3).print();

        /**
         * reduce 算子 使用演示
         * 需求： 求各性别中，好友数量最大的那个人，而且如果前后两个人的好友数量相同，则输出的结果中，也需要将uid/name等信息更新成后面一条数据的值
         *
         */
        source
                .map((MapFunction<String, UserInfo>) e -> JSONObject.parseObject(e, UserInfo.class))
                .map(bean -> Tuple4.of(bean.getUid(), bean.getName(), bean.getGender(), bean.getFriends().size()))
                .returns(new TypeHint<Tuple4<Integer, String, String, Integer>>() {})
                .keyBy(tp -> tp.f2)
                .reduce(new ReduceFunction<Tuple4<Integer, String, String, Integer>>() {
                    /**
                     *
                     * @param value1  是此前的聚合结果
                     * @param value2  是本次的新数据
                     * @return 更新后的聚合结果
                     * @throws Exception
                     */
                    @Override
                    public Tuple4<Integer, String, String, Integer> reduce(Tuple4<Integer, String, String, Integer> value1, Tuple4<Integer, String, String, Integer> value2) throws Exception {

                        if (value1 == null || value2.f3 >= value1.f3) {
                            return value2;
                        } else {
                            return value1;
                        }
                    }
                }).print();


        env.execute();


    }


    static class UserInfo implements Serializable {
        private int uid;
        private String name;
        private String gender;
        private List<FriendInfo> friends;

        public int getUid() {
            return uid;
        }

        public void setUid(int uid) {
            this.uid = uid;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public String getGender() {
            return gender;
        }

        public void setGender(String gender) {
            this.gender = gender;
        }

        public List<FriendInfo> getFriends() {
            return friends;
        }

        public void setFriends(List<FriendInfo> friends) {
            this.friends = friends;
        }
    }

    static class FriendInfo implements Serializable {
        private int fid;
        private String name;

        public int getFid() {
            return fid;
        }

        public void setFid(int fid) {
            this.fid = fid;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }
    }

    static class UserFriendInfo implements Serializable {
        private int uid;
        private String name;
        private String gender;
        private int fid;
        private String fname;

        public int getUid() {
            return uid;
        }

        public void setUid(int uid) {
            this.uid = uid;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public String getGender() {
            return gender;
        }

        public void setGender(String gender) {
            this.gender = gender;
        }

        public int getFid() {
            return fid;
        }

        public void setFid(int fid) {
            this.fid = fid;
        }

        public String getFname() {
            return fname;
        }

        public void setFname(String fname) {
            this.fname = fname;
        }

        @Override
        public String toString() {
            return "UserFriendInfo{" +
                    "uid=" + uid +
                    ", name='" + name + '\'' +
                    ", gender='" + gender + '\'' +
                    ", fid=" + fid +
                    ", fname='" + fname + '\'' +
                    '}';
        }
    }

}
